package com.project.MessageRoute.executor;

import com.project.MessageRoute.model.MessageRouteBean;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.TreeMap;

@Component
public class MessageConsumerExecutor {

    @Value("${message.route.group}")
    private String groupName;

    @Value("${message.route.groups:}")
    private String groups;

    private Map<String, PushConsumer> consumers = new TreeMap<>();

    @Autowired
    private PushBinder pushBinder;

    public PushConsumer bind(String name) {
        if (StringUtils.isBlank(name)) {
            name = groupName;
        }

        if (consumers.containsKey(name)) {
            return consumers.get(name);
        }

        PushConsumer consumer = pushBinder.bind(name);
        Map<String, List<MessageRouteBean>> beans = MessageRouteSetting.getBeans();
        try
        {
            System.out.println("===============>items: " + beans.entrySet().size());
            for (Map.Entry<String, List<MessageRouteBean>> map : beans.entrySet()) {
                for (MessageRouteBean bean : map.getValue()){
                    String grp_id = StringUtils.isBlank(bean.getRouteMethod().groupId()) ? groupName : bean.getRouteMethod().groupId();
                    if (name.equals(grp_id)) {
                        System.out.println(String.format("======================>bind topic: %s; tags: %s",
                                bean.getRouteMethod().topic(), bean.getRouteMethod().tags()));
                        consumer.subscribe(bean.getRouteMethod().topic(), bean.getRouteMethod().tags());
                    }
                }
            }

            consumer.start();
            consumers.put(name, consumer);
            System.out.println("===============>success: " + name);
        } catch (Exception e){
            System.out.println("===============>error: " + name);
            e.printStackTrace();
        }
        return consumer;
    }

    @Bean
    public PushConsumer messageConsumer(){
        bind(groupName);
        if (StringUtils.isNotBlank(groups)) {
            String[] group_list = groups.split(",");
            for (String s : group_list) {
                bind(s);
            }
        }

        return consumers.get(groupName);
    }

}
